package com.frinder.akka;

import akka.actor.ActorRef;
import akka.actor.Terminated;
import akka.actor.UntypedActor;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.routing.ActorRefRoutee;
import akka.routing.Routee;
import akka.routing.Router;
import akka.routing.SmallestMailboxRoutingLogic;
import com.google.common.collect.Lists;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * Created by frinder on 2017/7/26.
 */
@Component
@Scope("prototype")
public class MasterActor extends UntypedActor {

    private final LoggingAdapter logger = Logging.getLogger(getContext().system(), this);

    @Autowired
    private SpringExtension springExtension;

    /**
     * 路由器
     */
    private Router router = new Router(new SmallestMailboxRoutingLogic());


    @Override
    public void preStart() throws Exception {
        logger.info("{}", "preStart");
        List<Routee> routees = Lists.newArrayList();
        ActorRef actor;
        for (int i = 0; i < 100; i++) {
            actor = getContext().actorOf(springExtension.props("workerActor", 10));
            getContext().watch(actor);
//            routees.add(new ActorRefRoutee(actor));
            router = router.addRoutee(actor);
        }
//        router = new Router(new SmallestMailboxRoutingLogic(), routees);
        logger.info("router routees : {}", router.routees().size());
        super.preStart();
    }

    @Override
    public void onReceive(Object message) throws Throwable {
        if (message instanceof AkkaResponse) {
            logger.info("Response : {}", message);
        } else if (message instanceof AkkaRequest) {
            logger.debug("Redirect message to WorkerActor.");
            router.route(message, getSelf());
        } else if (message instanceof Terminated) {
            router = router.removeRoutee(((Terminated) message).actor());
            ActorRef actor = getContext().actorOf(springExtension.props("workerActor"));
            getContext().watch(actor);
            router = router.addRoutee(new ActorRefRoutee(actor));
        } else {
            logger.error("Error : {}", message);
            unhandled(message);
        }
    }
}
